use crate::cache::IndexChange;
use crate::{
    cache, field_db, log_db, message_db, model_db, run_by_store, sqlx_run_by_store, ROCKSDB,
};
use rust_rocksdb::WriteBatchWithTransaction;
use std::sync::Arc;
use visit_dd_core::model::DataStoreEngine;
use visit_dd_core::proj::{VddLogPart, VddMessage, APP_CONFIG};
use visit_dd_core::query::VddValueEvaluation;
use visit_dd_core::visit::{ShareRuleOfField, ShareRuleOfModel, VisitGroup, VisitRequest};
use visit_dd_core::{alog, i18n, Ares, Astr, UserId};

trait VisitDBTrait {
    fn from_vr(
        vr: &VisitRequest,
        limits: &[Arc<ShareRuleOfField>],
        filters: &[Arc<VddValueEvaluation>],
        rights: &[Arc<ShareRuleOfModel>],
    ) -> Self;
}

impl VisitDBTrait for VisitGroup {
    fn from_vr(
        vr: &VisitRequest,
        limits: &[Arc<ShareRuleOfField>],
        filters: &[Arc<VddValueEvaluation>],
        rights: &[Arc<ShareRuleOfModel>],
    ) -> Self {
        let mut vg_limits = Vec::from(limits);
        if !vr.write_able {
            // 生成所有字段都只读的规则
            let field_ids = field_db::impl_rdb::f_select_field_ids(vr.model_id);
            let limit = ShareRuleOfField::SrfRo(field_ids);
            vg_limits.push(Arc::new(limit));
        }
        VisitGroup {
            model_id: vr.model_id,
            group_type: vr.vgt.clone(),
            write_able: vr.write_able,
            rule_m: if rights.is_empty() {
                vr.rights.clone()
            } else {
                Vec::from(rights)
            },
            u_id: vr.request_uid.clone(),
            rule_f: vg_limits,
            rule_fv: Vec::from(filters),
        }
    }
}

pub async fn vr_save(vr: &VisitRequest, operator: &UserId, is_update: bool) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::vr_save(&mut batch, vr)
    } else {
        sqlx_run_by_store!(impl_sqlx::vr_save, vr)?
    };
    if !is_update {
        log_db::save(
            &mut batch,
            &mut index_changes,
            &alog!(
                VddLogPart::LpAction(vr.request_uid.to_owned(), "发送访问请求".into()),
                VddLogPart::LpModel(vr.model_id),
            ),
        );
        let model = model_db::m_load(vr.model_id).await;
        let txt = i18n!(1348, &model.name.clone());
        message_db::save(
            &mut batch,
            &mut index_changes,
            &VddMessage::new(&vr.request_uid, &model.uid, &txt),
        );
    } else {
        log_db::save(
            &mut batch,
            &mut index_changes,
            &alog!(
                if vr.passed() {
                    VddLogPart::LpAction(operator.to_owned(), "通过访问请求".into())
                } else {
                    VddLogPart::LpAction(operator.to_owned(), "拒绝访问请求".into())
                },
                VddLogPart::LpUser(vr.request_uid.to_owned()),
                VddLogPart::LpModel(vr.model_id),
            ),
        );
        message_db::save(
            &mut batch,
            &mut index_changes,
            &VddMessage::share_request_handle(operator.to_owned(), vr),
        );
    }
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?
    }
    ares
}

pub async fn vr_try_load_in_model_by_user(
    model_id: u64,
    request_uid: &UserId,
) -> Option<Arc<VisitRequest>> {
    run_by_store!(
        impl_rdb::vr_try_load_in_model_by_user,
        impl_sqlx::vr_try_load_in_model_by_user,
        model_id,
        request_uid
    )
}

pub async fn vr_select_in_model(model_id: u64) -> Vec<Arc<VisitRequest>> {
    run_by_store!(
        impl_rdb::vr_select_in_model,
        impl_sqlx::vr_select_in_model,
        model_id
    )
}

/// 批量获取一批模型的访问组计数
pub async fn vg_count_in_models(model_ids: &[u64]) -> Vec<usize> {
    run_by_store!(impl_rdb::count_vg, impl_sqlx::vg_count_in_models, model_ids)
}

pub async fn vr_count_in_model(model_id: u64) -> Result<(usize, usize), Astr> {
    run_by_store!(
        impl_rdb::vr_count_in_model,
        impl_sqlx::vr_count_in_model,
        model_id
    )
}

pub async fn vr_delete(model_id: u64, uid: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::vr_delete(&mut batch, model_id, uid)
    } else {
        sqlx_run_by_store!(impl_sqlx::vr_delete, model_id, uid)?
    };
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(uid.to_owned(), "取消访问请求".into()),
            VddLogPart::LpModel(model_id),
        ),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?
    }
    ares
}

pub async fn vg_save(vg: &VisitGroup, operator: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::vg_save(&mut batch, vg)
    } else {
        sqlx_run_by_store!(impl_sqlx::vg_save, vg)?
    };
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(operator.clone(), "保存".into()),
            VddLogPart::LpModel(vg.model_id),
            VddLogPart::LpVisitGroup(vg.model_id, vg.u_id.0)
        ),
    );
    message_db::save(
        &mut batch,
        &mut index_changes,
        &VddMessage::new(&operator, &vg.u_id, "开通了访问组".into()),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn vg_load(model_id: u64, u_id: UserId) -> Arc<VisitGroup> {
    run_by_store!(impl_rdb::vg_load, impl_sqlx::vg_load, model_id, u_id)
}

pub async fn vg_select_all() -> Vec<Arc<VisitGroup>> {
    run_by_store!(impl_rdb::vg_select_all, impl_sqlx::vg_select_all)
}

pub async fn vg_select_in_model(model_id: u64) -> Vec<Arc<VisitGroup>> {
    run_by_store!(
        impl_rdb::vg_select_in_model,
        impl_sqlx::vg_select_in_model,
        model_id
    )
}

pub async fn vg_close(vg: &VisitGroup, operator: &UserId) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    if APP_CONFIG.meta_store.is_none() {
        impl_rdb::vg_close(&mut batch, vg)
    } else {
        sqlx_run_by_store!(impl_sqlx::vg_close, vg)?
    };
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(operator.clone(), "关闭访问组".into()),
            VddLogPart::LpModel(vg.model_id),
            VddLogPart::LpVisitGroup(vg.model_id, vg.u_id.0),
        ),
    );
    message_db::save(
        &mut batch,
        &mut index_changes,
        &VddMessage::new(&operator, &vg.u_id, "关闭了访问组".into()),
    );
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

pub async fn vg_select_by_user_id(user_id: &UserId) -> Vec<Arc<VisitGroup>> {
    run_by_store!(
        impl_rdb::vg_select_by_user_id,
        impl_sqlx::vg_select_by_user_id,
        user_id
    )
}

pub async fn vg_from_vr(
    vr: &VisitRequest,
    limits: &[Arc<ShareRuleOfField>],
    filters: &[Arc<VddValueEvaluation>],
    rights: &[Arc<ShareRuleOfModel>],
) -> Result<u64, Astr> {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let r = if APP_CONFIG.meta_store.is_none() {
        impl_rdb::vg_from_vr(&mut batch, vr, limits, filters, rights)
    } else {
        sqlx_run_by_store!(impl_sqlx::vg_from_vr, vr, limits, filters, rights)
    };
    ROCKSDB.write(batch).map_err(|e| Arc::from(e.to_string()))?;
    r
}

pub async fn vg_try_load_in_model_by_user(
    model_id: u64,
    request_uid: &UserId,
) -> Option<Arc<VisitGroup>> {
    run_by_store!(
        impl_rdb::vg_try_load_in_model_by_user,
        impl_sqlx::vg_try_load_in_model_by_user,
        model_id,
        request_uid
    )
}

mod impl_rdb {
    use crate::count_in_key_head;
    use crate::keys::VddKey;
    use crate::visit_db::VisitDBTrait;
    use crate::{load_by_key_head, ROCKSDB};
    use rust_rocksdb::WriteBatchWithTransaction;
    use serde_json::json;
    use std::sync::Arc;
    use visit_dd_core::query::VddValueEvaluation;
    use visit_dd_core::visit::{ShareRuleOfField, ShareRuleOfModel, VisitGroup, VisitRequest};
    use visit_dd_core::{i18n, Astr, UserId, EMPTY_UID};

    pub(super) fn vr_save(batch: &mut WriteBatchWithTransaction<true>, vr: &VisitRequest) {
        let vk = VddKey::BoundVisitRequest(vr.model_id, vr.request_uid.clone());
        batch.put_cf(vk.db_name().unwrap(), vk.to_string(), json!(vr).to_string())
    }

    pub(super) fn vr_try_load_in_model_by_user(
        model_id: u64,
        request_uid: &UserId,
    ) -> Option<Arc<VisitRequest>> {
        let key = VddKey::BoundVisitRequest(model_id, request_uid.clone()).to_string();
        ROCKSDB
            .get(key)
            .map(|x| x.map(|y| Arc::new(serde_json::from_slice::<VisitRequest>(&y).unwrap())))
            .unwrap()
            .or(None)
    }

    pub(super) fn vr_select_in_model(model_id: u64) -> Vec<Arc<VisitRequest>> {
        let mut r = Vec::<Arc<VisitRequest>>::new();
        let kh = VddKey::BoundVisitRequest(model_id, EMPTY_UID);
        load_by_key_head(&kh).iter().for_each(|x| {
            let request = serde_json::from_slice::<VisitRequest>(x).unwrap();
            r.push(Arc::new(request));
        });
        r
    }

    pub(super) fn vr_count_in_model(model_id: u64) -> Result<(usize, usize), Astr> {
        let kh = VddKey::BoundVisitRequest(model_id, EMPTY_UID);
        let mut sum = 0;
        let mut handled = 0;
        load_by_key_head(&kh).iter().for_each(|x| {
            sum += 1;
            let request = serde_json::from_slice::<VisitRequest>(x).unwrap();
            if request.accept_ts.is_some() {
                handled += 1;
            }
        });
        Ok((sum, handled))
    }

    pub(super) fn vr_delete(
        batch: &mut WriteBatchWithTransaction<true>,
        model_id: u64,
        uid: &UserId,
    ) {
        let key = VddKey::BoundVisitRequest(model_id, uid.clone()).to_string();
        batch.delete(key)
    }

    pub(super) fn vg_save(batch: &mut WriteBatchWithTransaction<true>, vg: &VisitGroup) {
        let vk = VddKey::BoundVisitGroup(vg.model_id, vg.u_id.clone());
        batch.put_cf(vk.db_name().unwrap(), vk.to_string(), json!(vg).to_string())
    }

    /// 查看一批模型的访问组数目
    pub(super) fn count_vg(model_ids: &[u64]) -> Vec<usize> {
        let mut res = Vec::<usize>::with_capacity(model_ids.len());
        for model_id in model_ids {
            let kh = VddKey::BoundVisitGroup(*model_id, EMPTY_UID);
            let count = count_in_key_head(&kh, |_| true);
            res.push(count);
        }
        res
    }

    pub(super) fn vg_load(model_id: u64, u_id: UserId) -> Arc<VisitGroup> {
        let key = VddKey::BoundVisitGroup(model_id, u_id.clone()).to_string();
        let vg = ROCKSDB
            .get(&key)
            .map(|value| value.map(|v| serde_json::from_slice::<VisitGroup>(&v).unwrap()))
            .unwrap()
            .unwrap();
        Arc::new(vg)
    }

    pub(super) fn vg_select_all() -> Vec<Arc<VisitGroup>> {
        let key_head = VddKey::BoundVisitGroup(0, EMPTY_UID);
        let mut r: Vec<Arc<VisitGroup>> = Vec::new();
        load_by_key_head(&key_head).iter().for_each(|x| {
            let group = serde_json::from_slice::<VisitGroup>(x).unwrap();
            r.push(Arc::new(group));
        });
        r
    }

    pub(super) fn vg_select_in_model(model_id: u64) -> Vec<Arc<VisitGroup>> {
        let key_head = VddKey::BoundVisitGroup(model_id, EMPTY_UID);
        load_by_key_head(&key_head)
            .iter()
            .map(|x| {
                let vg = serde_json::from_slice::<VisitGroup>(x).unwrap();
                Arc::new(vg)
            })
            .collect::<Vec<Arc<VisitGroup>>>()
    }

    pub(super) fn vg_close(batch: &mut WriteBatchWithTransaction<true>, vg: &VisitGroup) {
        batch.delete(VddKey::BoundVisitGroup(vg.model_id, vg.u_id.clone()).to_string())
    }

    pub(super) fn vg_select_by_user_id(user_id: &UserId) -> Vec<Arc<VisitGroup>> {
        let kh = VddKey::BoundVisitGroup(0, user_id.clone());
        load_by_key_head(&kh)
            .iter()
            .map(|x| {
                let group = serde_json::from_slice::<VisitGroup>(x).unwrap();
                if group.u_id == user_id.clone() {
                    Some(Arc::new(group))
                } else {
                    None
                }
            })
            .filter(|x| x.is_some())
            .map(|x| x.unwrap())
            .collect::<Vec<Arc<VisitGroup>>>()
    }

    pub(super) fn vg_from_vr(
        batch: &mut WriteBatchWithTransaction<true>,
        vr: &VisitRequest,
        limits: &[Arc<ShareRuleOfField>],
        filters: &[Arc<VddValueEvaluation>],
        rights: &[Arc<ShareRuleOfModel>],
    ) -> Result<u64, Astr> {
        if vg_try_load_in_model_by_user(vr.model_id, &vr.request_uid).is_some() {
            return Err(i18n!(1344));
        }
        let vg = VisitGroup::from_vr(vr, limits, filters, rights);
        vg_save(batch, &vg);
        Ok(vg.u_id.0)
    }

    pub(super) fn vg_try_load_in_model_by_user(
        model_id: u64,
        request_uid: &UserId,
    ) -> Option<Arc<VisitGroup>> {
        let groups = vg_select_in_model(model_id);
        // 先看和用户匹配的
        let vg = groups
            .iter()
            .filter(|g| g.u_id == request_uid.clone())
            .map(|g| g.to_owned())
            .next();
        if vg.is_none() {
            // 如果没有和用户匹配的，看是否有允许匿名访问的
            groups
                .iter()
                .filter(|vg| vg.accept_anonymouse())
                .map(|vg| vg.to_owned())
                .next()
        } else {
            vg
        }
    }
}

mod impl_sqlx {
    use crate::keys::VddKey;
    use crate::sqlx::SqlxDefineTrait;
    use crate::visit_db::VisitDBTrait;
    use std::sync::Arc;
    use visit_dd_core::query::VddValueEvaluation;
    use visit_dd_core::visit::{ShareRuleOfField, ShareRuleOfModel, VisitGroup, VisitRequest};
    use visit_dd_core::EMPTY_UID;
    use visit_dd_core::{Ares, Astr, UserId};

    pub(super) async fn vr_save(pool: &impl SqlxDefineTrait, vr: &VisitRequest) -> Ares {
        pool.save_bound(
            VddKey::BoundVisitRequest(vr.model_id, vr.request_uid.clone()),
            &serde_json::to_string(vr).unwrap(),
        )
        .await
    }

    pub(super) async fn vr_try_load_in_model_by_user(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
        request_uid: &UserId,
    ) -> Option<Arc<VisitRequest>> {
        let bounds = pool
            .select_bounds(VddKey::BoundVisitRequest(model_id, request_uid.clone()))
            .await;
        bounds
            .first()
            .map(|x| Arc::new(serde_json::from_str(x).unwrap()))
    }

    pub(super) async fn vr_select_in_model(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
    ) -> Vec<Arc<VisitRequest>> {
        let bounds = pool
            .select_bounds(VddKey::BoundVisitRequest(model_id, EMPTY_UID))
            .await;
        bounds
            .iter()
            .map(|x| {
                let vr = serde_json::from_str::<VisitRequest>(&x).unwrap();
                Arc::new(vr)
            })
            .collect::<Vec<Arc<VisitRequest>>>()
    }

    pub(super) async fn vr_count_in_model(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
    ) -> Result<(usize, usize), Astr> {
        let sql_r = pool
            .select_bounds(VddKey::BoundVisitRequest(model_id, EMPTY_UID))
            .await;
        let mut sum = 0;
        let mut handled = 0;
        sql_r.iter().for_each(|x| {
            let vr = serde_json::from_str::<VisitRequest>(&x).unwrap();
            sum += 1;
            if vr.accept_ts.is_some() {
                handled += 1;
            }
        });
        Ok((sum, handled))
    }

    pub(super) async fn vg_count_in_models(
        pool: &impl SqlxDefineTrait,
        model_ids: &[u64],
    ) -> Vec<usize> {
        let bts = model_ids
            .iter()
            .map(|x| VddKey::BoundVisitGroup(*x, EMPTY_UID))
            .collect::<Vec<VddKey>>();
        let count_map = pool.count_bounds_grouped(&bts).await;
        model_ids
            .iter()
            .map(|x| match count_map.get(x) {
                Some(count) => *count,
                None => 0,
            })
            .collect::<Vec<usize>>()
    }

    pub(super) async fn vr_delete(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
        uid: &UserId,
    ) -> Ares {
        pool.delete_bounds(VddKey::BoundVisitRequest(model_id, uid.clone()))
            .await
    }

    pub(super) async fn vg_save(pool: &impl SqlxDefineTrait, vg: &VisitGroup) -> Ares {
        pool.save_bound(
            VddKey::BoundVisitGroup(vg.model_id, vg.u_id.clone()),
            &serde_json::to_string(&vg).unwrap(),
        )
        .await
    }

    pub(super) async fn vg_load(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
        u_id: UserId,
    ) -> Arc<VisitGroup> {
        let contents = pool
            .select_bounds(VddKey::BoundVisitGroup(model_id, u_id.clone()))
            .await;
        let vg = serde_json::from_str::<VisitGroup>(&contents.first().unwrap()).unwrap();
        Arc::new(vg)
    }

    pub(super) async fn vg_select_all(pool: &impl SqlxDefineTrait) -> Vec<Arc<VisitGroup>> {
        let bounds = pool
            .select_bounds(VddKey::BoundVisitGroup(0, EMPTY_UID))
            .await;
        bounds
            .iter()
            .map(|x| {
                let vg = serde_json::from_str::<VisitGroup>(x).unwrap();
                Arc::new(vg)
            })
            .collect::<Vec<Arc<VisitGroup>>>()
    }

    pub(super) async fn vg_select_in_model(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
    ) -> Vec<Arc<VisitGroup>> {
        let bounds = pool
            .select_bounds(VddKey::BoundVisitGroup(model_id, EMPTY_UID))
            .await;
        bounds
            .iter()
            .map(|x| {
                let vg = serde_json::from_str::<VisitGroup>(x).unwrap();
                Arc::new(vg)
            })
            .collect::<Vec<Arc<VisitGroup>>>()
    }

    pub(super) async fn vg_close(pool: &impl SqlxDefineTrait, vg: &VisitGroup) -> Ares {
        let _ = pool
            .delete_bounds(VddKey::BoundVisitGroup(vg.model_id, vg.u_id.clone()))
            .await;
        pool.delete_bounds(VddKey::BoundVisitRequest(vg.model_id, vg.u_id.clone()))
            .await
    }

    pub(super) async fn vg_select_by_user_id(
        pool: &impl SqlxDefineTrait,
        user_id: &UserId,
    ) -> Vec<Arc<VisitGroup>> {
        let mut bounds = pool
            .select_bounds(VddKey::BoundVisitGroup(0, EMPTY_UID)) // 因为要把匿名许可的访问凭证也查出来，所以不能传用户ID进去
            .await
            .iter()
            .map(|x| {
                let vg = serde_json::from_str::<VisitGroup>(x).unwrap();
                Arc::new(vg)
            })
            .collect::<Vec<Arc<VisitGroup>>>();
        bounds.retain(|bound| &bound.u_id == user_id || bound.accept_anonymouse());
        bounds
    }

    pub(super) async fn vg_from_vr(
        pool: &impl SqlxDefineTrait,
        vr: &VisitRequest,
        limits: &[Arc<ShareRuleOfField>],
        filters: &[Arc<VddValueEvaluation>],
        rights: &[Arc<ShareRuleOfModel>],
    ) -> Result<u64, Astr> {
        let vg = VisitGroup::from_vr(vr, limits, filters, rights);
        vg_save(pool, &vg).await?;
        Ok(vg.u_id.0)
    }

    pub(super) async fn vg_try_load_in_model_by_user(
        pool: &impl SqlxDefineTrait,
        model_id: u64,
        request_uid: &UserId,
    ) -> Option<Arc<VisitGroup>> {
        let bounds = pool
            .select_bounds(VddKey::BoundVisitGroup(model_id, EMPTY_UID)) // 因为要把匿名许可的访问凭证也查出来，所以不能传用户ID进去
            .await;
        let mut r: Option<Arc<VisitGroup>> = None;
        // 绑定关系是模型-访问组主键，用户信息需要解析访问组json后排查
        for x in bounds.iter() {
            let vg = serde_json::from_str::<VisitGroup>(x).unwrap();
            let vg_ref = Arc::new(vg);
            if &vg_ref.u_id == request_uid {
                r = Some(vg_ref.clone());
            }
            if vg_ref.accept_anonymouse() && r.is_none() {
                r = Some(vg_ref.clone());
            }
        }
        r
    }
}
